package com.adcc.monitoragent.core.service;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

import javax.management.JMX;
import javax.management.MBeanServerConnection;
import javax.management.ObjectName;
import javax.management.remote.JMXConnector;
import javax.management.remote.JMXConnectorFactory;
import javax.management.remote.JMXServiceURL;

import org.apache.activemq.broker.jmx.BrokerViewMBean;
import org.apache.activemq.broker.jmx.QueueViewMBean;
import org.apache.activemq.broker.jmx.TopicViewMBean;
import org.apache.activemq.web.RemoteJMXBrokerFacade;
import org.apache.activemq.web.config.SystemPropertiesConfiguration;
import org.quartz.DisallowConcurrentExecution;
import org.quartz.Job;
import org.quartz.JobExecutionContext;
import org.quartz.JobExecutionException;
import org.quartz.PersistJobDataAfterExecution;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.adcc.monitoragent.bean.MqBean;
import com.adcc.monitoragent.bean.MqBean.MqMetricsBean;
import com.adcc.monitoragent.log.ServiceLogger;
import com.adcc.monitoragent.mq.MqLink;
import com.adcc.monitoragent.send.MessageSend;
import com.adcc.monitoragent.util.Constants;
import com.adcc.monitoragent.util.ProtertiesUtil;
import com.tongtech.tlrt.model.MonitorBrokerInfo;
import com.tongtech.tlrt.model.MonitorQueueInfo;
import com.tongtech.tlrt.model.MonitorVersionInfo;
import com.tongtech.tlrt.monitor.proxy.MonitorProxy;

/**
 * ip + port 唯一确定 消息队列监控类，包括activeMq apolloMq等
 * 
 * @author charlie
 * 
 */
@PersistJobDataAfterExecution   
@DisallowConcurrentExecution
public class MqService implements Job {
	private final static Logger log = LoggerFactory.getLogger(MqService.class);
	private static Map<String, MqLink> testLinkMap = new ConcurrentHashMap<String, MqLink>();
	private static Map<String, RemoteJMXBrokerFacade> jmxMap = new ConcurrentHashMap<String, RemoteJMXBrokerFacade>();
	private static Map<String, MonitorProxy> mbeanProxyMap = new ConcurrentHashMap<String, MonitorProxy>();
	private static Map<String, JMXConnector> mbeanConnMap = new ConcurrentHashMap<String, JMXConnector>();
	private static Map<String, MqBean> lastMqBeanMap = new ConcurrentHashMap<String, MqBean>();

	/**
	 * 得到mq信息
	 * 支持activemq  和东方通 mq
	 * 东方通mq只支持queue 没有主题模式
	 * @param ip
	 * @param port
	 * @param userName
	 * @param passWard
	 * @param type
	 * @param mode
	 * @return
	 */
	public static MqBean getInfo(String connString, String testString, String clusterString, 
			String userName, String passWard, int maxMqQueueNumber) {
		MqBean mqBean = new MqBean();
		String[] connArray = connString.split(":");
		if(connArray.length == 3) {
			mqBean.setIp(connArray[0]);
			mqBean.setPort(connArray[1]);
			mqBean.setType(connArray[2]);
		}
		if(testString != null && !"".equals(testString)) {
			mqBean.setTestConnUrl("failover:(tcp://" + testString + ")?initialReconnectDelay=100&timeout=3000&jms.useAsyncSend=true");
		}
		if(clusterString != null && !"".equals(clusterString)) {
			String[] clusterArray = clusterString.split(":");
			mqBean.setIpCluster(clusterArray[0]);
			mqBean.setPortCluster(clusterArray[1]);
		}
		if (mqBean.getType().equals(MqBean.MQ_TYPE_ACTIVE)) {
			getActiveMQInfo(mqBean);
		} else if (mqBean.getType().equals(MqBean.MQ_TYPE_TLRT)) {
			getTLRTInfo(mqBean);
		}
		String key = mqBean.buildKey();
		MqBean lastMqBean = lastMqBeanMap.get(key);
		//计算单位时间队列传递数,  计算堆报，用户 上线，下线
		if(lastMqBean != null) {//启动后状态改变，发送日志
			//转换成map
			Map<String, MqMetricsBean> map = new HashMap<String, MqMetricsBean>();
			for (MqMetricsBean mqMetricsBean : lastMqBean.getMetrics()) {
				map.put(mqMetricsBean.buildKey(), mqMetricsBean);
			}
			//循环计算
			for (MqMetricsBean mqMetricsBean : mqBean.getMetrics()) {
				MqMetricsBean last = map.get(mqMetricsBean.buildKey());
				if(last != null) {
					//消费者数目是否改变
					int cons = mqMetricsBean.getConsumer().compareTo(last.getConsumer());
					if(cons > 0) {//上线
						//String desc = "Mq队列状态改变:用户上线";
						String desc = "用户上线";
						ServiceLogger.info(mqBean.getIp(), mqBean.getPort(), "queue-" + mqMetricsBean.getName(), desc);
					} else if (cons < 0) {//下线
						//String desc = "Mq队列状态改变:用户下线";
						String desc = "用户下线";
						ServiceLogger.error(mqBean.getIp(), mqBean.getPort(), "queue-" + mqMetricsBean.getName(), desc);
					}
					
					if(!mqMetricsBean.getConsumer().equals("0")) {//用户存在判断堆报
						long lastSize = Long.parseLong(last.getSize()) - maxMqQueueNumber;
						long nowSize =  Long.parseLong(mqMetricsBean.getSize()) - maxMqQueueNumber;
						long count = lastSize * nowSize;
						if(count < 0) {//状态不一致
							if(nowSize > 0) {//堆报
								//String desc = "Mq队列状态改变:用户堆报";
								String desc = "用户堆报";
								ServiceLogger.warn(mqBean.getIp(), mqBean.getPort(), "queue-" + mqMetricsBean.getName(), desc);
							} else {//堆报恢复
								//String desc = "Mq队列状态改变:用户堆报恢复";
								String desc = "用户堆报恢复";
								ServiceLogger.info(mqBean.getIp(), mqBean.getPort(), "queue-" + mqMetricsBean.getName(), desc);
							}
						}
					}
					
					//计算单位时间流量
					mqMetricsBean.setDequeuePer(Long.parseLong(mqMetricsBean.getDequeue()) -
							Long.parseLong(last.getDequeue()) + "");
				}
			}
			
		}
		
		return mqBean;
	}

	/**
	 * 得到东方通MQ信息
	 * 
	 * @param ip
	 * @param port
	 * @param type
	 * @param mode
	 * @param mqBean
	 */
	private static void getTLRTInfo(MqBean mqBean) {
		String urlStr = "service:jmx:rmi://" + mqBean.getIp() + "/jndi/rmi://" + mqBean.getIp() + ":11241/tlrt_monitor";
		try {
			JMXServiceURL url = new JMXServiceURL(urlStr);
			MonitorProxy mbeanProxy = mbeanProxyMap.get(urlStr);
			if (mbeanProxy == null) {
				//关闭历史连接
				JMXConnector conn = mbeanConnMap.remove(url.toString());
				if(conn != null) {
					conn.close();
				}
				//创建新的连接
				mbeanProxy = initTLRTJmx(url);
				mbeanProxyMap.put(urlStr, mbeanProxy);
			}
			MonitorBrokerInfo brokerInfo = mbeanProxy.queryBroker(mqBean.getIp(), Short
					.valueOf(mqBean.getPort()), 1);
			if(brokerInfo.getState() < 1){
				//RT 挂掉
				mqBean.setStatus(Constants.STATUS_ERROR);
			} else {
				mqBean.setId(String.valueOf(brokerInfo.getBrokerId()));
				MonitorVersionInfo versionInfo = mbeanProxy.queryVersion(mqBean.getIp(), Short
						.valueOf(mqBean.getPort()));
				mqBean.setVersion(String.valueOf(versionInfo.getProductinfo()));
				mqBean.setMetrics(getTLRTAllQueues(mqBean.getIp(), mqBean.getPort(),mbeanProxy));
				mqBean.setStatus(Constants.STATUS_OK);
			}
			
		} catch (Exception e) {
			log.error("获取TLRT状态失败", e);
			mbeanProxyMap.remove(urlStr);
			mqBean.setStatus(Constants.STATUS_ERROR);
		}
	}
	/**
	 * 获取东方通队列详细消息
	 * @param ip
	 * @param port
	 * @return
	 * @throws Exception
	 */
	public static List<MqMetricsBean> getTLRTAllQueues(String ip, String port, MonitorProxy mbeanProxy)
			throws Exception {
		List<MqMetricsBean> list = new ArrayList<MqMetricsBean>();
		List<MonitorQueueInfo> quelist = mbeanProxy.queryQueueList(ip, Short
				.valueOf(port));
		for (int i = 0; i < quelist.size(); i++) {
			MonitorQueueInfo q = quelist.get(i);

			MqMetricsBean metrics = new MqMetricsBean();
			metrics.setIp(ip);
			metrics.setName(q.getDestinationName());
			metrics.setConsumer(String.valueOf(q.getCurConsumerNum()));
			metrics.setDequeue(String.valueOf(q.getOutMsgNum()));
			metrics.setEnqueue(String.valueOf(q.getInMsgNum()));
			metrics.setProducer(String.valueOf(0));
			metrics.setSize(String.valueOf(q.getCurmsgnum()));

			list.add(metrics);
		}

		return list;
	}

	/**
	 * 初始化东方通MQ JMX链接
	 * 
	 * @param url
	 * @throws Exception
	 */
	private static MonitorProxy initTLRTJmx(JMXServiceURL url) throws Exception {
		JMXConnector jmxc = JMXConnectorFactory.connect(url, null);
		mbeanConnMap.put(url.toString(), jmxc);
		MBeanServerConnection mbsc = jmxc.getMBeanServerConnection();
		ObjectName mbeanName;
		mbeanName = new ObjectName(
				"com.tongtech.tlrt.monitor.proxy:name=MonitorProxy");
		return JMX.newMBeanProxy(mbsc, mbeanName, MonitorProxy.class,
				true);
	}

	private static void getActiveMQInfo(MqBean mqBean) {
		String url = "service:jmx:rmi:///jndi/rmi://" + mqBean.getIp() + ":" + mqBean.getPort()
				+ "/jmxrmi";

		try {
			RemoteJMXBrokerFacade jmx = jmxMap.get(url);
			if (jmx == null) {
				jmx = initJmx(url);
				jmxMap.put(url, jmx);
			}
			BrokerViewMBean broker = jmx.getBrokerAdmin();
			mqBean.setId(broker.getBrokerId());
			mqBean.setVersion(broker.getBrokerVersion());
			List<MqMetricsBean> list = new ArrayList<MqMetricsBean>();
			getAllQueues(broker, list, jmx, mqBean);
			getAllTopics(broker, list, jmx, mqBean);
			mqBean.setMetrics(list);
			//进行mq收发，并对session进行缓存
			MqLink link = testLinkMap.get(mqBean.getTestConnUrl());
			if(link == null) {
				link = new MqLink(mqBean.getTestConnUrl());
				testLinkMap.put(mqBean.getTestConnUrl(), link);
			}
			boolean testResult = link.doTest();
			if(testResult) {
				mqBean.setStatus(Constants.STATUS_OK);//链路收发正常
			} else {
				mqBean.setStatus(Constants.STATUS_ERROR);//链路收发异常
			}
		} catch (Exception e) {
			log.error("获取ActiveMQ状态失败" + e.getMessage());
			mqBean.setStatus(Constants.STATUS_ERROR);
			jmxMap.remove(url);
		}
	}
	/**
	 * 获取所有队列
	 * @param broker
	 * @param list
	 * @return
	 * @throws Exception
	 */
	public static List<MqMetricsBean> getAllQueues(BrokerViewMBean broker, List<MqMetricsBean> list, RemoteJMXBrokerFacade jmx, MqBean mqBean)
			throws Exception {
		ObjectName[] queuesQuenues = broker.getQueues(); // 队列
		List<String> queues = getNames(queuesQuenues);

		for (int i = 0; i < queues.size(); i++) {
			QueueViewMBean q = jmx.getQueue(queues.get(i));

			MqMetricsBean metrics = new MqMetricsBean();
			metrics.setIp(mqBean.getIp());
			metrics.setName(q.getName());
			metrics.setConsumer(String.valueOf(q.getConsumerCount()));
			metrics.setDequeue(String.valueOf(q.getDequeueCount()));
			metrics.setEnqueue(String.valueOf(q.getEnqueueCount()));
			metrics.setProducer(String.valueOf(q.getProducerCount()));
			metrics.setSize(String.valueOf(q.getQueueSize()));

			list.add(metrics);
		}

		return list;
	}
	/**
	 * 获取所有主题监控
	 * @param broker
	 * @param list
	 * @return
	 * @throws Exception
	 */
	private static List<MqMetricsBean> getAllTopics(BrokerViewMBean broker, List<MqMetricsBean> list, RemoteJMXBrokerFacade jmx, MqBean mqBean)
			throws Exception {
		ObjectName[] queuesTopics = broker.getTopics(); // 队列
		List<String> queues = getNames(queuesTopics);

		for (int i = 0; i < queues.size(); i++) {
			//无用队列过滤
			String name = queues.get(i);
			if(name == null || name.startsWith("ActiveMQ.Advisory.")) {
				continue;
			}
			TopicViewMBean t = jmx.getTopic(name);
			if(t == null || t.getName() == null) {
				continue;
			}
			MqMetricsBean metrics = new MqMetricsBean();
			metrics.setIp(mqBean.getIp());
			metrics.setName(t.getName());
			metrics.setConsumer(String.valueOf(t.getConsumerCount()));
			metrics.setDequeue(String.valueOf(t.getDequeueCount()));
			metrics.setEnqueue(String.valueOf(t.getEnqueueCount()));
			metrics.setProducer(String.valueOf(t.getProducerCount()));
			metrics.setSize(String.valueOf(t.getQueueSize()));
			metrics.setType(MqBean.MQ_MODE_TOPIC);
			list.add(metrics);
		}
		return list;
	}

	private static List<String> getNames(ObjectName[] queues) {
		List<String> list = new ArrayList<String>();

		for (int i = 0; i < queues.length; i++) {
			String name = queues[i].getCanonicalName();
			list.add(name.substring(name.indexOf("Destination=") + 12, name
					.indexOf(",Type")));
		}

		return list;
	}

	private static RemoteJMXBrokerFacade initJmx(String url) {
		RemoteJMXBrokerFacade jmx = new RemoteJMXBrokerFacade();
		System.setProperty("webconsole.jmx.url", url);
		SystemPropertiesConfiguration configuration = new SystemPropertiesConfiguration();
		jmx.setConfiguration(configuration);
		return jmx;
	}
	
	@Override
	public void execute(JobExecutionContext arg0) throws JobExecutionException {
		log.debug("定时执行任务。。。。。。。。。。。。");

		for (int i = 0; i < ProtertiesUtil.getMqConnectString().size(); i++) {
			String connString = ProtertiesUtil.getMqConnectString().get(i).toString();
			try {
				String testString = ProtertiesUtil.getMqTestString().size() > i ? ProtertiesUtil.getMqTestString().get(i).toString() : null;
				String clusterString = ProtertiesUtil.getMqClusterString().size() > i ? ProtertiesUtil.getMqClusterString().get(i).toString() : null;
				MqBean mqBean = MqService.getInfo(connString, testString, clusterString,
						ProtertiesUtil.getMqUsername(), ProtertiesUtil.getMqPassword(), ProtertiesUtil.getMaxMqQueueNumber());
				
				String key = mqBean.buildKey();
				MqBean lastMqBean = lastMqBeanMap.get(key);
				boolean statusChange = lastMqBean != null && !lastMqBean.getStatus().equals(mqBean.getStatus());
				if(mqBean.getStatus().equals(Constants.STATUS_ERROR) && statusChange) {//如果状态异常 并且 状态改变
					log.warn("Mq状态正常转异常，休眠2秒重试，防止误判");
					try {
						Thread.sleep(2000);
					} catch (InterruptedException e) {
					}
					mqBean = MqService.getInfo(connString, testString, clusterString,
							ProtertiesUtil.getMqUsername(), ProtertiesUtil.getMqPassword(), ProtertiesUtil.getMaxMqQueueNumber());
					if(mqBean.getStatus().equals(Constants.STATUS_ERROR)) {//状态还是异常
						//String desc = "Mq状态改变:" + lastMqBean.getStatus() + "->" + mqBean.getStatus();
						String desc = "服务停止";
						ServiceLogger.error(mqBean.getIp(), mqBean.getPort(), "mq", desc);
					}
				} else if(statusChange){//状态由异常转正常
					//String desc = "Mq状态改变:" + lastMqBean.getStatus() + "->" + mqBean.getStatus();
					String desc = "服务启动";
					ServiceLogger.info(mqBean.getIp(), mqBean.getPort(), "mq", desc);
				}
				lastMqBeanMap.put(key, mqBean);
				//转发消息
				MessageSend.sendMsg(mqBean);
			} catch (Exception e) {
				log.error("Mq定时任务执行失败:" + connString +  e.getMessage());
			}
		}
		log.debug("完成任务。。。。。。。。。。。。");
	}

	public static void main(String[] args) throws Exception {

	}
}
